package com.hzmg.route;

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.Props;
import akka.actor.typed.SupervisorStrategy;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.PoolRouter;
import akka.actor.typed.javadsl.Routers;
import com.hzmg.handler.RouteDefinitionHandler;
import com.hzmg.support.util.NameUtils;
import com.typesafe.config.Config;


public class PoolRouteConversion implements RouteConversion {
    private static PoolRouteConversion instance = new PoolRouteConversion();

    private static final String POOLROUTER_NAME = "poolRouter";

    private PoolRouteConversion() {
    }

    public static PoolRouteConversion getInstance() {
        return instance;
    }

    @Override
    public Behavior<?> routingBehavior(RouteDefinition routeDefinition) {
        return Behaviors.setup(
                context -> {
                    context.getLog().info("[akkacloud-core-gateway] PoolRouteConversion: {}"
                            , context.getSystem().settings().config().getConfig("akka.actor.deployment"));
                    String instancesConfig = context
                            .getSystem()
                            .settings()
                            .config()
                            .getConfig(NameUtils.ROUTE_CONFIGURATION_PREFIX + "." + POOLROUTER_NAME)
                            .getString("nr-of-instances");
                    int poolSize = Integer.valueOf(instancesConfig);

                    PoolRouter<Shift> pool =
                            Routers.pool(poolSize
                                            , Behaviors.supervise(RouteDefinitionHandler.create()).onFailure(SupervisorStrategy.restart()))
                                    .withRoundRobinRouting();

                    ActorRef<Shift> router = context.spawn(pool, POOLROUTER_NAME);
                    router.tell(routeDefinition);

                    return Behaviors.empty();
                }
        );
    }


}
